package com.gagakuai.bpm.config.mq;

import com.alibaba.fastjson.JSON;
import com.gagakuai.bpm.common.constants.RabbitMqConstants;
import lombok.Data;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/*
 * @program: mss_server
 * @description: RabbitMQ配置
 * @author: houhong
 * @create: 2022-03-24 14:31
 */
@Data
@Configuration
@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitConfig {


    private String host;

    private String port;

    private String username;

    private String password;

    private String virtualhost = "/";

    @Bean
    public ConnectionFactory connectionFactory() {

        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setVirtualHost("/");
        cachingConnectionFactory.setUsername(username);
        cachingConnectionFactory.setPassword(password);
        cachingConnectionFactory.setHost(host);
        cachingConnectionFactory.setPublisherConfirms(true);

        // 设置ack为true
        cachingConnectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        cachingConnectionFactory.setPublisherReturns(true);
        return cachingConnectionFactory;
    }


    /*
     * 创建一个bean，使用的消费者的消息确认机制默认为手动 是推模式
     *
     * @return
     */
    @Bean
    public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory() {

        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        // 配置rabbit每次发送指定数量的消息给对应的消费者
        factory.setPrefetchCount(1);

        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {

        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());

       /* rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            // 第一个参数: 是生产消息是传入的CorrelationData对象，里面维护了一个id，
            // 可以自定义取值来标识某些业务
            // 第二个参数: 判断消息有没有发成功
            // 第三个参数: 发生异常的原因，cause为异常的原因
            System.out.println("ack: " + ack);
            System.out.println("cause: " + cause);
            System.out.println("correlationData: " + correlationData);
        });*/

        // 允许失败回调
        rabbitTemplate.setMandatory(true);
      /*  rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            // 第一个参数: message -> 消息主体
            // 第二个参数: 发送失败错误码
            // 第三个参数: 发送失败错误信息
            // 第四个参数: 发送的交换机名字
            // 第五个参数: 发送消息的routingKey
            System.out.println(message);
            System.out.println(replyCode);
            System.out.println(replyText);
            System.out.println(exchange);
            System.out.println(routingKey);
        });*/

        // 自己实现了一个消息转换器
        rabbitTemplate.setMessageConverter(new MessageConverter() {
            @Override
            public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
                // 指定发送的消息类型为test/plain
                messageProperties.setContentType("text/plain");
                // 指定消息的编码格式
                messageProperties.setContentEncoding("UTF-8");
                Message message = new Message(JSON.toJSONBytes(object), messageProperties);
                return message;
            }

            @Override
            public Object fromMessage(Message message) throws MessageConversionException {
                return message;
            }
        });

        return rabbitTemplate;
    }


    /*
     * 备用交换机配置:使用广播交换机
     *
     * @return
     */
    @Bean
    public FanoutExchange defaultExchange() {

        return (FanoutExchange) ExchangeBuilder.
                fanoutExchange(RabbitMqConstants.DEFAULT_EXCHANGE).build();
    }

    /*
     * 创建一个topic交换机 附带一个备胎
     */
    @Bean
    public TopicExchange topicExchange() {

        Map<String, Object> map = new HashMap<>();
        map.put(RabbitMqConstants.ALTER_EXCHANGE_DEFINE, RabbitMqConstants.DEFAULT_EXCHANGE);

        return (TopicExchange) ExchangeBuilder
                .topicExchange(RabbitMqConstants.BPM_TOPIC_EXCHANGE)
                .withArguments(map).build();
    }

    /*
     * bpm撤销流程队列
     */
    @Bean(RabbitMqConstants.BPM_WITH_DRAW_QUEUE_NAME)
    public Queue queue() {
        // durable: 是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive: 默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete: 是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        Map<String, Object> deadQueueArgs = deadQueueArgs();
        return new Queue(RabbitMqConstants.BPM_WITH_DRAW_QUEUE_NAME, true, false, false, deadQueueArgs);
    }

    /*
     * bpm取回流程队列
     */
    @Bean(RabbitMqConstants.BPM_WITH_RETURN_QUEUE_NAME)
    public Queue withReturn() {

        Map<String, Object> deadQueueArgs = deadQueueArgs();
        return new Queue(RabbitMqConstants.BPM_WITH_RETURN_QUEUE_NAME, true, false, false, deadQueueArgs);
    }

    /*
     * bpm撤销到上一步队列
     */
    @Bean(RabbitMqConstants.BPM_WITH_DRAW_LAST_STEP_QUEUE_NAME)
    public Queue withDrawLastStepQueue() {

        Map<String, Object> deadQueueArgs = deadQueueArgs();
        return new Queue(RabbitMqConstants.BPM_WITH_DRAW_LAST_STEP_QUEUE_NAME, true, false, false, deadQueueArgs);
    }

    /*
     * bpm撤销到第一步队列
     */
    @Bean(RabbitMqConstants.BPM_WITH_DRAW_FIRST_STEP_QUEUE_NAME)
    public Queue withDrawFirstStepQueue() {

        Map<String, Object> deadQueueArgs = deadQueueArgs();
        return new Queue(RabbitMqConstants.BPM_WITH_DRAW_FIRST_STEP_QUEUE_NAME, true, false, false, deadQueueArgs);
    }


    /*
     * bpm 同步人员信息队列
     */
    @Bean(RabbitMqConstants.BPM_WITH_SYNC_ACCOUNT_QUEUE_NAME)
    public Queue syncAccountQueue() {

        return new Queue(RabbitMqConstants.BPM_WITH_SYNC_ACCOUNT_QUEUE_NAME);
    }

    /*
     * 绑定
     */
    @Bean
    public Binding binding() {
        // 将队列和routing key绑定至exchange中
        return BindingBuilder
                .bind(queue())
                .to(topicExchange())
                .with(RabbitMqConstants.BPM_WITH_DRAW_SERVICE_MATCH_ROUTING_KEY);
    }

    /*
     * 绑定
     */
    @Bean
    public Binding bindingLastStepQueue() {
        // 将队列和routing key绑定至exchange中
        return BindingBuilder
                .bind(withDrawLastStepQueue())
                .to(topicExchange())
                .with(RabbitMqConstants.BPM_WITH_DRAW_LAST_STEP_SERVICE_MATCH_ROUTING_KEY);
    }


    /*
     * 绑定
     */
    @Bean
    public Binding bindingFirstStepQueue() {
        // 将队列和routing key绑定至exchange中
        return BindingBuilder
                .bind(withDrawFirstStepQueue())
                .to(topicExchange())
                .with(RabbitMqConstants.BPM_WITH_DRAW_FIRST_STEP_SERVICE_MATCH_ROUTING_KEY);
    }

    /*
     * 绑定取回队列
     */
    @Bean
    public Binding bindingWithReturnQueue() {
        // 将队列和routing key绑定至exchange中
        return BindingBuilder
                .bind(withReturn())
                .to(topicExchange())
                .with(RabbitMqConstants.BPM_WITH_RETURN_FIRST_STEP_SERVICE_MATCH_ROUTING_KEY);
    }

    /*
     * 绑定
     */
    @Bean
    public Binding bindingSyncAccountQueue() {
        // 将队列和routing key绑定至exchange中
        return BindingBuilder
                .bind(syncAccountQueue())
                .to(topicExchange())
                .with(RabbitMqConstants.BPM_SYNC_ACCOUNT_MATCH_ROUTING_KEY);
    }

    /*  业务提示队列bpm服务ACK*****************************************************/
    /*
     * bpm回调队列
     */
    @Bean(RabbitMqConstants.BPM_CALLBACK_ACK_QUEUE_NAME)
    public Queue bpmAckQueue() {

        return new Queue(RabbitMqConstants.BPM_CALLBACK_ACK_QUEUE_NAME, true, false, false);
    }

    /*
     * 绑定bpm异常回调数据
     */
    @Bean
    public Binding bindingAckBackQueue() {
        // 将队列和routing key绑定至exchange中
        return BindingBuilder
                .bind(bpmAckQueue())
                .to(topicExchange())
                .with(RabbitMqConstants.BPM_CALLBACK_ACK_ROUTKING_KEY);
    }


    /* 业务系统回调 bpm服务的错误数据，进行回滚业务状态*****************************************************/

    /*
     * bpm回调队列
     */
    @Bean(RabbitMqConstants.BPM_EXCEPTION_CALLBACK_QUEUE_NAME)
    public Queue bpmCallBackQueue() {

        Map<String, Object> callBackdeadQueueArgs = callBackdeadQueueArgs();
        return new Queue(RabbitMqConstants.BPM_EXCEPTION_CALLBACK_QUEUE_NAME, true, false, false, callBackdeadQueueArgs);
    }

    /*
     * 绑定bpm异常回调数据
     */
    @Bean
    public Binding bindingBpmCallBackQueue() {
        // 将队列和routing key绑定至exchange中
        return BindingBuilder
                .bind(bpmCallBackQueue())
                .to(topicExchange())
                .with(RabbitMqConstants.BPM_EXCEPTION_CALLBACK_ROUTKING_KEY);
    }

    /* 死信配置 *****************************************************/

    /*
     * 死信交换机
     */
    @Bean
    DirectExchange deadExchange() {
        return new DirectExchange(RabbitMqConstants.BPM_MSG_DEAD_EXCHANGE, true, false);
    }

    /*
     * 死信队列
     */
    @Bean(name = RabbitMqConstants.BPM_DEAD_QUEUE_NAME)
    public Queue deadQueue() {
        return new Queue(RabbitMqConstants.BPM_DEAD_QUEUE_NAME, true, false, false);
    }

    /*
     * 死信队列绑定
     */
    @Bean
    Binding deadRouteBinding() {
        return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(RabbitMqConstants.BPM_DEAD_QUEUE_MATCH_ROUTING_KEY);
    }

    /*
     * 转发到 死信队列，配置参数
     */
    private Map<String, Object> deadQueueArgs() {

        Map<String, Object> map = new HashMap<>();
        // 绑定该队列到私信交换机
        map.put(RabbitMqConstants.DEAD_LETTER_QUEUE_KEY, RabbitMqConstants.BPM_MSG_DEAD_EXCHANGE);
        //绑定路由key
        map.put(RabbitMqConstants.DEAD_LETTER_ROUTE_KEY, RabbitMqConstants.BPM_DEAD_QUEUE_MATCH_ROUTING_KEY);
        return map;
    }

    /* 回调死信配置 *****************************************************/

    /*
     * 回调死信交换机
     */
    @Bean(name = RabbitMqConstants.BPM_CALLBACK_DEAD_EXCHANGE)
    DirectExchange callBackDeadExchange() {
        return new DirectExchange(RabbitMqConstants.BPM_CALLBACK_DEAD_EXCHANGE, true, false);
    }

    /*
     * 回调死信队列
     */
    @Bean(name = RabbitMqConstants.BPM_CALLBACK_DEAD_QUEUE_NAME)
    public Queue callBackDeadQueue() {
        return new Queue(RabbitMqConstants.BPM_CALLBACK_DEAD_QUEUE_NAME, true, false, false);
    }

    /*
     * 死信队列绑定
     */
    @Bean
    Binding callbackDeadRouteBinding() {
        return BindingBuilder.bind(callBackDeadQueue()).to(callBackDeadExchange()).with(RabbitMqConstants.BPM_CALLBACK_DEAD_QUEUE_MATCH_ROUTING_KEY);
    }

    /*
     * 转发到 死信队列，配置参数
     */
    private Map<String, Object> callBackdeadQueueArgs() {

        Map<String, Object> map = new HashMap<>();
        // 绑定该队列到私信交换机
        map.put(RabbitMqConstants.DEAD_LETTER_QUEUE_KEY, RabbitMqConstants.BPM_CALLBACK_DEAD_EXCHANGE);
        //绑定路由key
        map.put(RabbitMqConstants.DEAD_LETTER_ROUTE_KEY, RabbitMqConstants.BPM_CALLBACK_DEAD_QUEUE_MATCH_ROUTING_KEY);
        return map;
    }

}